package com.snail.learn.roketmq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class ProducerDemo {

    public static void main(String[] args) {
//        try {
//            sendSyncMessage();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }

//        try {
//            sendAsyncMessage();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }

//        try {
//            sendOneWayMessage();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }

        try {
            sendSeqMessage();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 发送同步消息:会等待消息发送结果，确保消息的可靠性
     */
    public static void sendSyncMessage() throws Exception {
        DefaultMQProducer producer = startProducer();
        for (int i = 0; i < 10; i++) {
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            Message msg = createMsg("demoTopic", "syncTag", "TestMsg" + i);
            msg.setDelayTimeLevel(2);
            // 5.发送消息
            msg.putUserProperty("order",i+"");
            SendResult sendResult = producer.send(msg);

            System.out.printf("%s%n", sendResult);
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
        // 6.关闭生产者producer
    }

    /**
     * 发送异步消息:不会等待消息发送结果，但可以有回调函数监控是否发送成功
     */
    public static void sendAsyncMessage() throws Exception {
        DefaultMQProducer producer = startProducer();
        for (int i = 0; i < 10; i++) {
            //        4.创建消息对象，指定主题Topic、Tag和消息体
            Message msg = createMsg("demoTopic", "asyncTag", "TestMsg" + i);
            //        5.发送消息
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("发送结果：%s",sendResult);
                }
                @Override
                public void onException(Throwable throwable) {
                    System.out.printf("发送异常：%s",throwable);
                }
            });
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
        //6.关闭生产者producer
    }


    /**
     * 发送单向消息
     */
    public static void sendOneWayMessage() throws Exception {
        DefaultMQProducer producer = startProducer();
        for (int i = 0; i < 10; i++) {
            //        4.创建消息对象，指定主题Topic、Tag和消息体
            Message msg = createMsg("demoTopic", "oneWayTag", "TestMsg" + i);
            //        5.发送消息
            producer.sendOneway(msg);
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
//        6.关闭生产者producer
    }


    /**
     * 发送顺序消息
     */
    public static void sendSeqMessage() throws Exception {
        DefaultMQProducer producer = startProducer();
        for (int i = 0; i < 10; i++) {
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            Message msg = createMsg("demoTopic", "seqTag", "TestMsg" + i);
            // 5.发送消息
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    // o 即为选择队列的业务标识，即传递过来的i（实际可以传递业务的标识，如id等）
                    return list.get((int) o);
                }
            }, i%4);// 选择队列的业务标识
            System.out.printf("%s%n", sendResult);
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
        //6.关闭生产者producer
    }

    /**
     * 发送事务消息：先发半消息，再处理本地事务，再依据本地事务的处理结果来决定是提交消息，还是撤回消息；亦或是不等待回查
     */
    public static void sendTranMessage() throws Exception {
        TransactionMQProducer producer = startTranProducer();
        for (int i = 0; i < 10; i++) {
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            Message msg = createMsg("demoTopic", "syncTag", "TestMsg" + i);
            // 5.发送消息
            SendResult sendResult = producer.sendMessageInTransaction(msg,null);

            System.out.printf("%s%n", sendResult);
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
        // 6.关闭生产者producer
    }

    private static Message createMsg(String topic, String tag, String msg) throws UnsupportedEncodingException {
        return new Message(topic,
                tag,
                msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
    }

    private static DefaultMQProducer startProducer() throws MQClientException {
        //        1.创建生产者producer，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("snailTestGroup");
//        2.指定Nameserver地址
        producer.setNamesrvAddr("localhost:9876");
//        3.启动producer
        producer.start();
        return producer;
    }

    private static TransactionMQProducer startTranProducer() throws MQClientException {
        // 1.创建生产者producer，并指定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("snailTestGroup");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("localhost:9876");

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                // 执行本地事务
                if(StringUtils.equals("Tag1",message.getTags())){
                    return LocalTransactionState.COMMIT_MESSAGE; // 本地事务提交，消息提交
                }
                if(StringUtils.equals("Tag2",message.getTags())){
                    return LocalTransactionState.ROLLBACK_MESSAGE; // 本地事务回滚，消息撤回
                }
                if(StringUtils.equals("Tag3",message.getTags())){
                    return LocalTransactionState.UNKNOW; // 本地事务不响应，消息待处理（回查），
                }
                return LocalTransactionState.UNKNOW;
            }
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                // 消息回查处理逻辑（确定本地事务处理结果）。
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        // 3.启动producer
        producer.start();
        return producer;
    }
}
